package com.vip.mvc.akka;

import akka.actor.ActorRef;
import akka.actor.ActorSystem;
import akka.actor.Props;
import akka.actor.UntypedActor;
import akka.dispatch.Futures;
import akka.dispatch.Mapper;
import akka.dispatch.OnFailure;
import akka.dispatch.OnSuccess;
import akka.event.Logging;
import akka.event.LoggingAdapter;
import akka.pattern.Patterns;
import akka.util.Timeout;
import com.vip.mvc.utils.JSONUtils;
import io.netty.buffer.Unpooled;
import io.netty.util.CharsetUtil;
import org.junit.Test;
import scala.concurrent.Await;
import scala.concurrent.Future;
import scala.concurrent.duration.Duration;

import scala.concurrent.Promise;

import java.io.IOException;
import java.util.concurrent.Callable;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;

import static junit.framework.TestCase.assertTrue;
import static org.junit.Assert.assertEquals;


/**
 * Created by cloudy.xie on 2015/1/8.
 */
public class HelloFuture {
    ActorSystem system = ActorSystem.create("mySystem");
    private final Duration timeout = Duration.create(50, TimeUnit.SECONDS);


    @Test
    public void testFuture(){
        final int k=1000;
        Futures.future(() -> {
            testK(k);
            return null;
        }, system.dispatcher());
    }

    private void testK(int k){
        k++;
        System.out.println("k="+k);
    }

    @Test
    public void mustBeAbleToExecuteAnOnResultCallback() throws Throwable {

        System.out.println("thread1="+Thread.currentThread().getName());

        final CountDownLatch latch = new CountDownLatch(1);
//        Promise<String> cf = Futures.promise();
//        Future<String> f = cf.future();
//        f.onSuccess(new OnSuccess<String>() {
//            public void onSuccess(String result) {
//                if (result.equals("foo")) {
//                    latch.countDown();
//                }
//                System.out.println("thread2="+Thread.currentThread().getName());
//            }
//        }, system.dispatcher());
//
//        cf.success("foo");

        Future<String> f = Futures.future(new Callable<String>() {
            public String call() throws InterruptedException {
                Thread.sleep(5000);
                return "Hello" + "World";
            }
        }, system.dispatcher());

//        // system.dispatcher()返回一个ExecutionContextExecutor
//        // 当异步任务执行成功时，打印执行结果
//        f.onSuccess(new OnSuccess<String>() {
//            public void onSuccess(String result) {
//                if (result.equals("foo")) {
//                    latch.countDown();
//                }
//            }
//        }, system.dispatcher());

        String resp = Await.result(f, timeout);
        System.out.println("aaa resp = "+resp);
//        assertTrue(latch.await(5, TimeUnit.SECONDS));
//        assertEquals(resp, "foo");
//    }
//
//    @Test
//    public void mustBeAbleToExecuteAnOnExceptionCallback() throws Throwable {
        System.out.println("thread3="+Thread.currentThread().getName());
        final CountDownLatch latch2 = new CountDownLatch(1);
        Promise<String> cf2 = Futures.promise();
        Future<String> f2 = cf2.future();

        f2.onFailure(new OnFailure() {
            public void onFailure(Throwable t) {
                if (t instanceof NullPointerException)
                    latch2.countDown();
                System.out.println("thread4="+Thread.currentThread().getName());
            }
        }, system.dispatcher());

        Throwable exception = new NullPointerException();
        cf2.failure(exception);
        assertTrue(latch2.await(5, TimeUnit.SECONDS));
        assertEquals(f2.value().get().failed().get(), exception);
    }

    @Test
    public void test(){
        String a = "123.32";
        String b = "10000.32";
        String c = "121.32";
        System.out.println(Double.valueOf(a).compareTo(Double.valueOf(b)));
        System.out.println(Double.valueOf(b).compareTo(Double.valueOf(a)));
        System.out.println(Double.valueOf(a).compareTo(Double.valueOf(c)));
        System.out.println(Double.valueOf(a).compareTo(Double.valueOf(a)));
    }

//    @Test
//    public void promiseTest() throws Throwable {
//        System.out.println("thread3="+Thread.currentThread().getName());
//        final CountDownLatch latch = new CountDownLatch(1);
//        Promise<String> cf = Futures.promise();
//
//        Future<String> f = cf.future();
//        f.flatMap(result->{
//            latch.countDown();
//            System.out.println("thread2="+Thread.currentThread().getName());
//            cf.success(result);
//            return cf.future();
//        });
//
//        Future<String> f = cf.future();
//        f.onFailure(new OnFailure() {
//            public void onFailure(Throwable t) {
//                if (t instanceof NullPointerException)
//                    latch.countDown();
//                System.out.println("thread4="+Thread.currentThread().getName());
//            }
//        }, system.dispatcher());
//
//        Throwable exception = new NullPointerException();
//        cf.failure(exception);
//        assertTrue(latch.await(5, TimeUnit.SECONDS));
//        assertEquals(f.value().get().failed().get(), exception);
//    }

    @Test
    public void mustBeAbleToFlatMapAFuture() throws Throwable {
        System.out.println("thread3="+Thread.currentThread().getName());
        final CountDownLatch latch = new CountDownLatch(1);
        Promise<String> cf = Futures.promise();
        cf.success("1000");
        Future<String> f = cf.future();
        Future<AAA> r = f.flatMap(new Mapper<String, Future<AAA>>() {
            public Future<AAA> checkedApply(String r) throws Throwable {
                if (false) throw new IOException("Just here to make sure this compiles.");
                latch.countDown();

                AAA aaa = new AAA();
                Promise<AAA> cf = Futures.promise();
                aaa.setA(Integer.parseInt(r));
                aaa.setB("test"+r);
                cf.success(aaa);
                System.out.println("thread4="+Thread.currentThread().getName());
                return cf.future();
            }
        }, system.dispatcher());

        AAA resp =  Await.result(r, timeout);
        System.out.println("aaa resp = "+resp);
        assertEquals(Await.result(f, timeout), "1000");
        assertEquals(resp.getA(), 1000);
        assertTrue(latch.await(50, TimeUnit.SECONDS));
    }

    class AAA{
        int a;
        String b;

        public int getA() {
            return a;
        }

        public void setA(int a) {
            this.a = a;
        }

        public String getB() {
            return b;
        }

        public void setB(String b) {
            this.b = b;
        }

        @Override
        public String toString() {
            try {
                return JSONUtils.toJSON(this);
            } catch (IOException e) {
                e.printStackTrace();
            }
            return null;
        }
    }


    public static class A extends UntypedActor {

        private final LoggingAdapter log = Logging.getLogger(getContext().system(), this);

        @Override
        public void onReceive(Object message) throws Exception {
            if (message instanceof String) {
                Thread.sleep(3000);
                System.out.println(Thread.currentThread().getName() + " sleep end");
                System.out.println("接收的消息：" + message);
                // 返回一个消息
                this.getSender().tell("result back hello world", this.getSelf());
                System.out.println("sender path=" + this.getSender().path());
                getContext().stop(this.getSelf());
                log.info("|||{} has stop", this.getSelf().path());
            }
        }
    }

    public static void main(String args[]) throws Exception {
        System.out.println(Thread.currentThread().getName());
        ActorSystem system = ActorSystem.create("mySystem");
        ActorRef a = system.actorOf(Props.create(A.class), "helloWorld1");
        Timeout timeout = new Timeout(Duration.create(5, "seconds"));
        Future<Object> future = Patterns.ask(a, "are you ready?", timeout);
        Future<Object> future2 = Patterns.ask(a, "are you ready333?", timeout);

        // This will cause the current thread to block and wait for the UntypedActor to ‘complete’
        // the Future with it’s reply.
        // 在这里会阻塞到 Await.result 方法上，但这会导致性能的损失。
        String result = (String) Await.result(future, timeout.duration());
        System.out.println("result="+result);
        String result2 = (String) Await.result(future2, timeout.duration());
        System.out.println("result2="+result2);
    }

}
